package com.own.component.elastic.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import com.own.component.common.base.entity.po.BaseBean;
import com.own.component.elastic.entity.BaseElasticPo;
import com.own.component.elastic.entity.BaseElasticQuery;
import com.own.component.elastic.entity.BaseElasticVo;
import com.own.component.elastic.exception.ElasticException;
import com.own.component.elastic.exception.ElasticExceptionEnum;
import com.own.component.elastic.service.BaseElasticService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * ABaseElasticService
 *
 * @author chenxueli
 * @date 2022-09-03 18:42:00
 */
@Slf4j
@Service
public abstract class AbstractBaseElasticService<
        PO extends BaseBean,
        ELASTIC_PO extends BaseElasticPo<PO>,
        ELASTIC_VO extends BaseElasticVo<ELASTIC_PO>,
        QUERY extends BaseElasticQuery
        >
        extends AbstractBaseElasticSearchService<PO, ELASTIC_PO, ELASTIC_VO, QUERY>
        implements BaseElasticService<PO, ELASTIC_PO, ELASTIC_VO, QUERY> {

    private static final Set<String> INDEX_SET = new HashSet<>();

    @Resource
    private ElasticsearchClient elasticsearchClient;

    /**
     * 初始化搜索引擎数据（主要是用来创建索引信息）
     */
    @Override
    public void init() {
        createIndex();
    }

    /**
     * 创建相关索引
     */
    @Override
    public final void createIndex() {
        try {
            elasticsearchClient.indices().create(builder ->
                    builder.index(index()).mappings(mappings -> mappings.properties(mapping()))
            );
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.INDEX_CREATE_ERROR, e);
        }
    }

    /**
     * 删除相关索引
     */
    @Override
    public void deleteIndex() {
        if (!checkIndex()) {
            return;
        }
        try {
            elasticsearchClient.indices().delete(builder -> builder.index(index()));
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.INDEX_DELETE_ERROR, e);
        }
    }

    /**
     * 验证相关索引
     */
    @Override
    public boolean checkIndex() {
        if (INDEX_SET.contains(index())) {
            return true;
        }
        try {
            var index = index();
            var response = elasticsearchClient.indices().exists(builder -> builder.index(index));
            var flag = response.value();
            if (flag) {
                INDEX_SET.add(index);
            }
            return flag;
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.INDEX_EXIST_ERROR, e);
        }
    }

    /**
     * 插入数据
     *
     * @param po 添加的对象
     */
    @Override
    public final void add(ELASTIC_PO po) {
        if (!checkIndex()) {
            createIndex();
        }
        try {
            elasticsearchClient.index(document -> document
                    .index(index())
                    .routing(routing(po))
                    .id(String.valueOf(po.getId()))
                    .document(po)
            );
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_SAVE_ERROR, e);
        }
    }

    /**
     * 批量插入数据
     *
     * @param list 列表数据
     */
    @Override
    public final void batchAdd(Collection<ELASTIC_PO> list) {
        var operations = list.stream().map(item ->
                BulkOperation.of(bulk ->
                        bulk.index(document -> document
                                .id(String.valueOf(item.getId()))
                                .document(item)
                                .routing(routing(item))
                        )
                )
        ).toList();
        bulk(operations);
    }

    /**
     * 修改数据
     *
     * @param id 主键id
     * @param po 添加的对象
     */
    @Override
    public final void update(String id, ELASTIC_PO po) {
        try {
            elasticsearchClient.update(document -> document.index(index()).id(id).doc(po), getElasticPoClass());
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_UPDATE_ERROR, e);
        }
    }

    /**
     * 修改或插入数据
     *
     * @param id 主键id
     * @param po 添加的对象
     */
    @Override
    public void upsert(String id, ELASTIC_PO po) {
        try {
            elasticsearchClient.update(document -> document.docAsUpsert(true).index(index()).id(id).doc(po).detectNoop(true), getElasticPoClass());
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_UPDATE_ERROR, e);
        }
    }

    /**
     * 批量修改或插入数据
     *
     * @param list 列表数据
     */
    @Override
    public void batchUpsert(Collection<ELASTIC_PO> list) {
        var operations = list.stream().map(item ->
                BulkOperation.of(bulk ->
                        bulk.update(document ->
                                document.id(String.valueOf(item.getId())).action(a -> a.docAsUpsert(true).doc(item))
                        )
                )
        ).toList();
        bulk(operations);
    }

    /**
     * 删除数据
     *
     * @param id 主键id
     */
    @Override
    public final void delete(String id) {
        try {
            // 确认是否存在原有数据
            var response = elasticsearchClient.exists(document -> document.index(index()).id(id));
            // 如果存在，删除数据信息
            if (response.value()) {
                elasticsearchClient.delete(document -> document.index(index()).id(id));
            }
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_DELETE_ERROR, e);
        }
    }

    /**
     * 批量删除数据
     *
     * @param idList 主键id列表
     */
    @Override
    public final void batchDelete(Collection<String> idList) {
        var operations = idList.stream().map(id -> BulkOperation.of(bulk ->
                        bulk.delete(document -> document.id(id))
                )
        ).toList();
        bulk(operations);
    }

    /**
     * 批量执行方法
     *
     * @param operations 操作列表
     */
    private void bulk(List<BulkOperation> operations) {
        try {
            var response = elasticsearchClient.bulk(x -> x.index(index()).operations(operations));
            if (response.errors()) {
                throw new ElasticException(ElasticExceptionEnum.DATA_SAVE_ERROR);
            }
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_SAVE_ERROR, e);
        }
    }

}
